package com.myrabbitmqdemo.demo.three;

import com.myrabbitmqdemo.demo.utils.RabbitMqUtils;
import com.myrabbitmqdemo.demo.utils.SleepUtils;
import com.rabbitmq.client.CancelCallback;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.DeliverCallback;

import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.Arrays;
import java.util.concurrent.TimeoutException;

/**
 *
 * 消费者：模拟实际开发中一个进程走的快，一个进程走的慢的场景
 * 消息在手动应答时是不丢失、放回队列中重新消费
 */
public class Work03 {

    // 队列名称
    public static final String TASK_QUEUE_NAME = "ack_queue";

    public static void main(String[] args) throws IOException, TimeoutException {
        Channel channel = RabbitMqUtils.getChannel();

        System.out.println("C1等待接收消息处理时间较短");
        DeliverCallback deliverCallback = (consumerTag,message) -> {
            // 沉睡1s
            SleepUtils.sleep(1);
            System.out.println("接收到的消息："+new String(message.getBody(), StandardCharsets.UTF_8));

            // 手动应答
            /**
             * 1、消息的标记 tag,每个队列都有一个属性，里面包含了这个标记
             * 2、是否批量应答，true是的，false不是
             */
            channel.basicAck(message.getEnvelope().getDeliveryTag(),false);
        };

        CancelCallback cancelCallback = (consumerTag) -> {
            System.out.println(consumerTag+":消息被中断");
        };

        boolean autoAck = false;

        // 不公平分发这个prefetchCount这个等于0是轮询分发，等于1是不公平分发，等于大于1的是预取值分发,
        // 在预处理分发的情况下prefetchCount这个是队列能留下的峰值，最多能有这么多的数据暂存在这个位置
//        int prefetchCount = 1;
        int prefetchCount = 2;
        //  不公平分发设置
        channel.basicQos(prefetchCount);

        channel.basicConsume(TASK_QUEUE_NAME,autoAck,deliverCallback,cancelCallback);

    }

}
